package com.wtw.recommender

import java.net.InetAddress

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest
import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.TransportAddress
import org.elasticsearch.transport.client.PreBuiltTransportClient

case class Movie(mid: Int, name: String, descri: String, timelong: String, issue: String,
                 shoot: String, language: String, genres: String, actors: String, directors: String)

case class Rating(uid: Int, mid: Int, score: Double, timestamp: Int)

case class Tag(uid: Int, mid: Int, tag: String, timestamp: Int)

case class MongoConfig(uri: String, db: String)

case class MysqlConfig(url: String, user: String, password: String, driver: String)

case class ESConfig(httpHosts: String, transportHosts: String, index: String, clustername: String)


object MyDataLoader {

  val MOVIE_DATA_PATH = "recommender/DataLoader/src/main/resources/movies.csv"
  val RATING_DATA_PATH = "recommender/DataLoader/src/main/resources/ratings.csv"
  val TAG_DATA_PATH = "recommender/DataLoader/src/main/resources/tags.csv"
  val MYSQL_MOVIE_COLLECTION = "Movie"
  val MYSQL_RATING_COLLECTION = "Rating"
  val MYSQL_TAG_COLLECTION = "Tag"
  val ES_MOVIE_INDEX = "Movie"

//  def storeDataInMongoDB(movieDF: DataFrame, ratingDF: DataFrame, tagDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit = {
//    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))
//
//    mongoClient(mongoConfig.db)(MYSQL_MOVIE_COLLECTION).dropCollection()
//    mongoClient(mongoConfig.db)(MYSQL_RATING_COLLECTION).dropCollection()
//    mongoClient(mongoConfig.db)(MYSQL_TAG_COLLECTION).dropCollection()
//
//    movieDF.write.option("uri", mongoConfig.uri).option("collection", MYSQL_MOVIE_COLLECTION)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save()
//
//    ratingDF.write.option("uri", mongoConfig.uri).option("collection", MYSQL_RATING_COLLECTION)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save()
//
//    tagDF.write.option("uri", mongoConfig.uri).option("collection", MYSQL_TAG_COLLECTION)
//      .mode("overwrite")
//      .format("com.mongodb.spark.sql")
//      .save()
//
//    mongoClient(mongoConfig.db)(MYSQL_MOVIE_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
//    mongoClient(mongoConfig.db)(MYSQL_RATING_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
//    mongoClient(mongoConfig.db)(MYSQL_RATING_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
//    mongoClient(mongoConfig.db)(MYSQL_TAG_COLLECTION).createIndex(MongoDBObject("uid" -> 1))
//    mongoClient(mongoConfig.db)(MYSQL_TAG_COLLECTION).createIndex(MongoDBObject("mid" -> 1))
//
//    mongoClient.close()
//  }

  def storeDataInES(movieDF: DataFrame)(implicit eSConfig: ESConfig): Unit = {
    val settings = Settings.builder().put("cluster.name", eSConfig.clustername).build()
    val esClient = new PreBuiltTransportClient(settings)

    val REGEX_HOST_PORT = "(.+):(\\d+)".r
    eSConfig.transportHosts.split(",").foreach{
      case REGEX_HOST_PORT(host:String,port:String) => {
        esClient.addTransportAddress(new
            TransportAddress(InetAddress.getByName(host),port.toInt))
      }
    }

    if(esClient.admin().indices().exists(new IndicesExistsRequest(eSConfig.index)).actionGet().isExists){
      esClient.admin().indices().delete(new DeleteIndexRequest(eSConfig.index))
    }

    esClient.admin().indices().create(new CreateIndexRequest(eSConfig.index))

    movieDF.write
      .option("es.nodes",eSConfig.httpHosts)
      .option("es.http.timeout","100m")
      .option("es.mapping.id","mid")
      .mode("overwrite")
      .format("org.elasticsearch.spark.sql")
      .save(eSConfig.index+"/"+ES_MOVIE_INDEX)

  }

  def storeDataInMysql(movieDF: DataFrame, ratingDF: DataFrame, tagDF: DataFrame)(implicit mysqlConfig: MysqlConfig) = {
    movieDF.write.mode("overwrite")
      .format("jdbc")
      .option("driver", mysqlConfig.driver)
      .option("url", mysqlConfig.url)
      .option("dbtable", MYSQL_MOVIE_COLLECTION) //表名
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .save()

    ratingDF.write.mode("overwrite")
      .format("jdbc")
      .option("driver", mysqlConfig.driver)
      .option("url", mysqlConfig.url)
      .option("dbtable", MYSQL_RATING_COLLECTION) //表名
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .save()

    tagDF.write.mode(SaveMode.Overwrite)
      .format("jdbc")
      .option("driver", mysqlConfig.driver)
      .option("url", mysqlConfig.url)
      .option("dbtable", MYSQL_TAG_COLLECTION) //表名
      .option("user", mysqlConfig.user)
      .option("password", mysqlConfig.password)
      .save()
  }

  def main(args: Array[String]): Unit = {

    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://192.168.120.201:27017/recommender",
      "mongo.db" -> "recommender",
      "mysql.url" -> "jdbc:mysql://localhost:3306/recommend",
      "mysql.user" -> "root",
      "mysql.password" -> "root",
      "mysql.driver" -> "com.mysql.jdbc.Driver",
      "es.httpHosts" -> "localhost:9200",
      "es.transportHosts" -> "localhost:9300",
      "es.index" -> "recommender",
      "es.cluster.name" -> "elasticsearch"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    val sc = spark.sparkContext

    import spark.implicits._

    val movieRDD = sc.textFile(MOVIE_DATA_PATH)
    val movieDF = movieRDD.map(x => {
      val attr = x.split("\\^")
      Movie(attr(0).toInt, attr(1).trim(), attr(2).trim(), attr(3).trim(), attr(4).trim(), attr(5).trim(),
        attr(6).trim(), attr(7).trim(), attr(8).trim(), attr(9).trim())
    }).toDF()

    val ratingRDD = sc.textFile(RATING_DATA_PATH)
    val ratingDF = ratingRDD.map(item => {
      val attr = item.split(",")
      Rating(attr(0).toInt,attr(1).toInt,attr(2).toDouble,attr(3).toInt)
    }).toDF()

    val tagRDD = spark.sparkContext.textFile(TAG_DATA_PATH)
    //将tagRDD装换为DataFrame
    val tagDF = tagRDD.map(item => {
      val attr = item.split(",")
      Tag(attr(0).toInt,attr(1).toInt,attr(2).trim,attr(3).toInt)
    }).toDF()

//    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))
//    storeDataInMongoDB(movieDF, ratingDF, tagDF)

    implicit val mysqlConfig = MysqlConfig(config("mysql.url"), config("mysql.user"), config("mysql.password"), config("mysql.driver"))
    storeDataInMysql(movieDF, ratingDF, tagDF)

    import org.apache.spark.sql.functions._
    val newTag = tagDF.groupBy($"mid")
      .agg(concat_ws("|",collect_set($"tag"))
        .as("tags"))
      .select("mid","tags")

    val movieWithTagsDF = movieDF.join(newTag, Seq("mid"), "left")

    implicit val esConfig = ESConfig(config("es.httpHosts"),
      config("es.transportHosts"),
      config("es.index"),
      config("es.cluster.name"))

    storeDataInES(movieWithTagsDF)

    spark.stop()

  }
}
